package com.gis507.mapreduce.design;

import com.gis507.mapreduce.AISBean.AisClassifyBean;
import com.gis507.mapreduce.AISBean.AnalysisIndex;
import com.gis507.mapreduce.Utils.UtilsClass;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class AISInterpolationMapper extends Mapper<LongWritable,Text, Text, AisClassifyBean> {

    private Text outK = new Text();
    private AisClassifyBean outV = new AisClassifyBean();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        String line = value.toString();

        String[] comment = line.split(",");

        int seaRange = UtilsClass.turnToInt(comment[13]);
        double lat_d = UtilsClass.turnToDouble(comment[12]);

        //String mmsi, String shipType, String nacStatusCN, double draught,
        // double heading, double course, double speed, String dest,
        // long unixTime, double lon_d, double lat_d, int seaRange
        if (seaRange == 2 && lat_d > 37.426 && lat_d < 41) {
            String mmsi = comment[0];
            String shipType = comment[1];
            String nacStatusCN = comment[2];
            double length = UtilsClass.turnToDouble(comment[3]);
            double width = UtilsClass.turnToDouble(comment[4]);
            double draught = UtilsClass.turnToDouble(comment[5]);
            double heading = UtilsClass.turnToDouble(comment[6]);
            double course = UtilsClass.turnToDouble(comment[7]);
            double speed = UtilsClass.turnToDouble(comment[8]);
            String dest = comment[9];
            long unixTime = UtilsClass.turnToLong(comment[10]);
            double lon_d = UtilsClass.turnToDouble(comment[11]);


            double tonnage = UtilsClass.turnToDouble(comment[14]);

            //fengzhuang
            outK.set(mmsi);

            outV.setMmsi(mmsi);
            outV.setShipType(shipType);
            outV.setNacStatusCN(nacStatusCN);
            outV.setLength(length);
            outV.setWidth(width);
            outV.setDraught(draught);
            outV.setHeading(heading);
            outV.setCourse(course);
            outV.setSpeed(speed);
            outV.setDest(dest);
            outV.setUnixTime(unixTime);
            outV.setLon_d(lon_d);
            outV.setLat_d(lat_d);
            outV.setTonnge(tonnage);

            //get the value of seaRange
            //        int seaRange = UtilsClass.getSeaRangeNumber(outV);

            //set seaRange for outV
            outV.setSeaRange(seaRange);

            //write for Mapper
            context.write(outK, outV);
        }
    }
}
